/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.neptune_test.common;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.aop.ClassInfoUtil;
import com.nhncorp.neptune.common.aop.ProxyFactory;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.ipc.NRPC;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.master.NeptuneMaster;
import com.nhncorp.neptune.master.NeptuneMasterIF;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tabletserver.TabletServerIF;

public class MiniNeptuneCluster implements Constants {
  private static final Logger LOG = Logger.getLogger(MiniNeptuneCluster.class.getName());
  
  protected NConfiguration conf;
//  protected MiniDFSCluster dfsCluster;
  protected NeptuneFileSystem fs;
  //protected NeptuneMasterIF master;
  protected Thread masterThread;
  protected Map<String, TabletServerIF> tabletServers;
  protected Map<String, Thread> tabletServerThreads;
  protected Map<String, NeptuneMasterIF> masterServers;
  protected Map<String, Thread> masterServerThreads;
  private GPath parentDir;
  private boolean format;
  
  public MiniNeptuneCluster(NConfiguration conf,
                            int nMasterServer, 
                            int nTabletServer, 
                            boolean format, 
                            Map<String, List<String>> handlers) throws IOException {
    this.conf = conf;
    this.tabletServers = new HashMap<String, TabletServerIF>();
    this.tabletServerThreads = new HashMap<String, Thread>();
    this.masterServers = new HashMap<String, NeptuneMasterIF>();
    this.masterServerThreads = new HashMap<String, Thread>();
    this.format = format;
    
    init(handlers, nMasterServer, nTabletServer);
  }
  
  private void init(Map<String, List<String>> handlers,
                      int nMasterServer, 
                      int nTabletServer) throws IOException {
                      //int handlerServerCount) throws IOException {
    this.fs = NeptuneFileSystem.get(conf);
    this.parentDir = new GPath(
        conf.get("neptune.root", 
            System.getProperty("user.home") + File.separator + ".neptune_root"));
    fs.mkdirs(parentDir);
    ClassInfoUtil.clearClassInfo();

    if(handlers != null) {
      for(Map.Entry<String, List<String>> entry: handlers.entrySet()) {
        String classId = entry.getKey();
        List<String> classHandlers = entry.getValue();
        ClassInfoUtil.getInstance().setInvokeHandler(classId, classHandlers);
      }
    }

    if(this.format) {
//      NeptuneMaster.formatFileSystem(conf);
      if(!NeptuneMaster.doFormat(conf)) {
        throw new IOException("Can't format");
      }
    }

    // Create the master
    startMasterServers(nMasterServer);

    startTabletServers(nTabletServer);
  }  
  
  private void startMasterServers(final int nMasterServer)
  throws IOException {
    for(int i = 0; i < nMasterServer; i++) {
      NConfiguration masterServerConf = new NConfiguration(conf);
      masterServerConf.set("masterServer.port", 9001 + i);
      masterServerConf.set("masterServer.handler.count", 4);
      masterServerConf.set("masterServer.heartbeat.port", 50200 + i);
      NeptuneMasterIF masterServer = (NeptuneMasterIF)ProxyFactory.createProxyClass("masterServer");
      masterServer.init(masterServer, masterServerConf);
      this.masterServers.put(masterServer.getHostName(), masterServer);
      Thread t = new Thread(masterServer, "MasterServer-" + this.masterServers.size());
      t.start();
      this.masterServerThreads.put(masterServer.getHostName(), t);
    }
  }
  
  private void startTabletServers(final int nTabletServer)
  throws IOException {
    for(int i = 0; i < nTabletServer; i++) {
      NConfiguration tabletServerConf = new NConfiguration(conf);
      tabletServerConf.set("tabletServer.port", 7001 + i);
      tabletServerConf.set("tabletServer.scanner.port", 50101 + i);
      tabletServerConf.set("tabletServer.handler.count", 4);
      tabletServerConf.set("tabletServer.heartbeat.port", 50300 + i);
      TabletServerIF tabletServer = (TabletServerIF)ProxyFactory.createProxyClass("tabletServer");
      tabletServer.init(tabletServer, tabletServerConf);
      this.tabletServers.put(tabletServer.getHostName(), tabletServer);
      Thread t = new Thread(tabletServer, "TabletServer-" + this.tabletServers.size());
      t.start();
      this.tabletServerThreads.put(tabletServer.getHostName(), t);
    }
  }

  public NeptuneMasterIF getActiveMaster() throws IOException {
    for(Map.Entry<String, NeptuneMasterIF> entry: masterServers.entrySet()) {
      if(entry.getValue().tryMasterLock()) {
        return entry.getValue();
      }
    }
    return null;
  }
  
  public void shutdown() throws IOException {

    //stopMasterServer();
    for(Map.Entry<String, NeptuneMasterIF> entry: masterServers.entrySet()) {
      entry.getValue().shutdown();
    }


    for(Thread masterServerThread: masterServerThreads.values()) {
      masterServerThread.interrupt();
    }
    
    for(Map.Entry<String, TabletServerIF> entry: tabletServers.entrySet()) {
      entry.getValue().shutdown();
    }


    for(Thread tabletServerThread: tabletServerThreads.values()) {
      tabletServerThread.interrupt();
    }
    
    NRPC.stopClient();
    try {
      Thread.sleep(2 * 1000);
    } catch (InterruptedException e) {
    }  
  }

  public void stopTabletServer(String hostName) throws IOException {
    System.out.println("Stop tablet server:" + hostName);
    TabletServerIF tabletServer = tabletServers.get(hostName);
    if(tabletServer == null) {
      throw new IOException("No live tablet");
    }
    TabletInfo[] tabletInfos = tabletServer.reportTablets();
    for(TabletInfo tabletInfo: tabletInfos) {
      System.out.println("\t stopping service: " + tabletInfo);
    }

    tabletServer.shutdown();
    tabletServers.remove(hostName);
    
    tabletServerThreads.get(hostName).interrupt();
  }  
  
  /**
   * 현재 active한 NeptuneMaster를 중지시킨다.
   * @throws IOException
   */
  public void stopMasterServer() throws IOException {
    LOG.debug("Active Master server shutdown");
    NeptuneMasterIF master = getActiveMaster();
    if(master != null) {
      master.shutdown();
      masterServerThreads.get(master.getHostName()).interrupt();
    }

    try {
      Thread.sleep(5 * 100);
    } catch (InterruptedException e) {
    }
  }
  
  public Map<String, TabletServerIF> getTabletServers() {
    return tabletServers;
  }

}
